package com.navinfo.opentsp.platform.computing.serializer;

import com.google.common.base.Charsets;

import com.navinfo.opentsp.platform.computing.service.AbstractProtocolSerializer;
import com.navinfo.opentsp.platform.computing.service.IProtocolParseService;
import com.navinfo.opentsp.platform.computing.util.InstanceUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;

public class HDFSCommonProtocolSerializer extends AbstractProtocolSerializer implements EventSerializer {

    private static final Logger logger = LoggerFactory.getLogger(HDFSCommonProtocolSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final OutputStream out;
    private final IProtocolParseService protocolParseService;

    private HDFSCommonProtocolSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
        String yard = ctx.getString("yard");
        String protocol = ctx.getString("protocol");
        logger.error("********************************************************");
        logger.error(Thread.currentThread().getName() + ", time is " + System.currentTimeMillis() + " , yard is " + yard + ", protocol is " + protocol);
        logger.error("********************************************************");
        this.protocolParseService = InstanceUtils.getInstanceService(yard, protocol);
    }

    public boolean supportsReopen() {
        return false;
    }

    public void afterCreate() {}

    public void afterReopen() {}

    public void beforeClose() {}

    public void flush() throws IOException {
        this.out.flush();
    }

    public void write(Event event) throws IOException {
        try {
            String data = protocolParseService.parseProtocol(event);
            //将0F39报文拼接后写入hdfs
            this.out.write(data.getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        }catch (Exception e){
            logger.error("执行HDFS的HDFSCommonProtocolSerializer拦截器失败：" + e.getMessage(), e);
        }
    }

    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSCommonProtocolSerializer(out, context);
        }
    }


}
